package com.heima.article.stream;


import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandler {


    //聚合流式处理
//        stream.map((key,value)->{
//            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
//
//            //重置消息的Key 和 value
//            return new KeyValue<>(mess.getArticleId().toString(),mess.getType().name()+":"+mess.getAdd());
//        }).groupBy((key,value)->key)
//                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
//                .aggregate(new Initializer<String>() {
//                    @Override
//                    public String apply() {
//
//                        return null;
//                    }
//                }, new Aggregator<String, String, String>() {
//                    @Override
//                    public String apply(String key, String value, String agg) {
//
//                        return null;
//                    }
//                },Materialized.as("hot-article-string-count-001")).toStream().to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){


        KStream<String,String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);

        stream.map((key,value)->{
            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
            return new KeyValue<>(mess.getArticleId().toString(),mess.getType().name()+":"+mess.getAdd());
        })
                .groupBy((key,value)->key).windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                .aggregate(new Initializer<String>() {
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                }, new Aggregator<String, String, String>() {
                    @Override
                    public String apply(String key, String value, String aggValue) {
                        String[] split = aggValue.split(",");
                        int col = 0,com = 0,lik = 0, vie = 0;
                        for (String agg : split) {
                            String[] split1 = agg.split(":");
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(split1[0])){
                                case COLLECTION:
                                    col = Integer.parseInt(split1[1]);
                                    break;
                                case LIKES:
                                    lik = Integer.parseInt(split1[1]);
                                    break;
                                case VIEWS:
                                    vie = Integer.parseInt(split1[1]);
                                    break;
                                case COMMENT:
                                    com = Integer.parseInt(split1[1]);
                                    break;
                            }
                        }
                        String[] split2 = value.split(":");
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(split2[0])){
                            case COLLECTION:
                                col += Integer.parseInt(split2[1]);
                                break;
                            case LIKES:
                                lik += Integer.parseInt(split2[1]);
                                break;
                            case VIEWS:
                                vie += Integer.parseInt(split2[1]);
                                break; 
                            case COMMENT:
                                com += Integer.parseInt(split2[1]);
                                break;
                        }
                        String format = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                        System.out.println("文章的id"+key);
                        System.out.println("当前消息窗口的数据处理结果"+format);

                        return format;
                    }
                },Materialized.as("hot-article-string-count-001"))
                .toStream()
                .map((key,value)->{
                    return new KeyValue<>(key.key().toString(),formatObj(key.key().toString(),value));
                    
                })
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);


        return stream;

    }

    private String formatObj(String toString, String value) {

        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.valueOf(toString));

        String[] valAry = value.split(",");
        for (String val : valAry) {
            String[] split = val.split(":");
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(split[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
            }
        }
        log.info("聚合消息处理后的结果为");
        return JSON.toJSONString(mess);
    }


}
